# -*- encoding: utf-8 -*-

"""
------------------------------------------
@File       : 62.thread_to_coroutine.py
@Author     : maixiaochai
@Email      : maixiaochai@outlook.com
@CreatedOn  : 2021/8/10 13:46
------------------------------------------
    将多份日志文件合并成一条日志输出流，以便我们调试程序

    给定日志文件句柄之后，我们得想办法判断有没有新数据的到来，如果有，就返回下一行输入内容。
    可以通过调用文件句柄的 tell方法来实现，并判断当前读取到的这个位置是否为未见中的最后一个位置。
    若是，则说明没有新数据，这时应该抛出异常

    这里的代码主要是演示 将任务从多线程一点点迁移到协程的过程，
    涉及到在线程中执行协程、在协程中启动线程并等待运行结果等内容
"""
import collections
from asyncio import get_event_loop, run_coroutine_threadsafe, gather, run, sleep as async_sleep, create_task
from threading import Lock, Thread
from time import sleep


class NoNewData(Exception):
    pass


def readline(handle):
    # tell() 方法返回文件的当前位置，即文件指针当前位置
    offset = handle.tell()

    # seek(offset[, whence]) 方法用于移动文件读取指针到指定位置
    # offset: 开始的偏移量，也就是代表需要移动偏移的字节数
    # whence：可选，默认值为 0。给offset参数一个定义，表示要从哪个位置开始偏移；0代表从文件开头开始算起，1代表从当前位置开始算起，2代表从文件末尾算起
    handle.seek(0, 2)

    length = handle.tell()

    # 读到文件最后的位置
    if length == offset:
        raise NoNewData

    # 这里如果不移动指针，默认从第一行开始读
    handle.seek(offset, 0)
    return handle.readline()


def tail_file(handle, interval, write_func):
    """
        一条工作线程，如果出现了新数据行，就通过用户传来的 write_fun回调函数把这行数据写到输出日志里面
    """
    while not handle.closed:
        try:
            line = readline(handle)
        except NoNewData:
            sleep(interval)

        # try没有出现异常时，else会执行
        else:
            write_func(line)


def run_thread(handles, interval, output_path):
    with open(output_path, 'wb') as output:
        lock = Lock()

        def write(data):
            with lock:
                output.write(data)

        threads = []

        for handle in handles:
            args = (handle, interval, write)
            thread = Thread(target=tail_file, args=args)
            thread.start()
            threads.append(thread)

        for thread in threads:
            # 主线程任务结束之后，进入阻塞状态，一直等待其他的子线程执行结束之后，主线程再终止
            thread.join()


def confirm_merge(input_paths, output_path):
    """
        判断输入文件是否正确地被合并了
    """
    found = collections.defaultdict(list)
    with open(output_path, 'rb') as f:
        for line in f:
            for path in input_paths:
                if line.find(path.encode()) == 0:
                    found[path].append(line)

    expected = collections.defaultdict(list)
    for path in input_paths:
        with open(path, 'rb') as f:
            expected[path].extend(f.readlines())

    for key, expected_lines in expected.items():
        found_lines = found[key]
        assert expected_lines == found_lines, \
            f'{expected_lines!r} == {found_lines!r}'


# input_paths = ...
# handles = ...
# output_path = ...

# print("input_paths: ", input_paths)
# print("input_paths type: ", type(input_paths))

# # ================================ [ 开始转换 ] ================================
"""
这里采用 “从上向下”的转换方向，从项目的最高点（例如main函数）开始，把这一部分改掉之后，
再沿着调用体系向下，去转换它所调用的那些函数与类。

具体步骤：
    1）修改当前的顶层函数，把声明方式从 def改为 async def
    2）把这个函数所要执行的 I/O调用（也就是有可能阻塞事件循环的那些调用）全部都用 asyncio.run_in_executor封装起来
    3）确保 run_in_executor所使用的资源以及它所触发的回调函数都经过适当的同步处理（这需要通过 Lock或 asyncio.run_coroutine_threadsafe
       函数来实现）
    4）沿着调用体系向下，按照前三个步骤将当前函数所调用到的其它函数与方法迁移到协程方案上面，看看这样迁移能不能把 get_event_loop和 run_in_executor
       从目前这一层里面拿掉
"""


# 1.第一次改进
async def run_tasks_mixed(handles, interval, output_path):
    """
        run_coroutine_threadsafe 函数提交写入操作，
        负责执行写入操作的 write_async协程，无论由哪一条工作线程提交，
        最终都会安排到事件循环（即loop）里执行（这个时间循环一般位于主线程中，如果有必要，也可以放在其它某条线程里面）。
        这些协程全部都是放在同一条先层里面执行的，因此，这实际上意味着，他们对输出文件所做的写入操作本身就会有秩序地执行，
        而不会出现相互重叠的情况

        只要 gather有了执行结果，我们就可以认定，对输出文件所做的那些写入操作全都已经执行完成了，
        于是，我们可以放心地让 with结构把表示输出文件的那个 output句柄关掉，而不用担心其中是否有写入操作还没执行完
    """
    loop = get_event_loop()

    with open(output_path, 'wb') as output:
        async def write_async(data):
            output.write(data)

        # write 不需要加锁，因为是线程安全的
        def write(data):
            coro = write_async(data)
            future = run_coroutine_threadsafe(coro, loop)
            future.result()

        tasks = []
        for handle in handles:
            # run_in_executor前不加 await，这样能够形成许许多多条并发的执行路径，从而实现 fan-out
            # 第一个参数指定 executor，如果是None则用默认的executor
            task = loop.run_in_executor(None, tail_file, handle, interval, write)
            tasks.append(task)

        await gather(*tasks)  # 收集任务的执行结果，从而实现 fan-in

    # 运行, asyncio.run
    # run(run_tasks_mixed(handles, 0.1, 'data/output.log'))

    # 按照第4步的方法，接下来修改 tail_async


# 第二次改进
async def tail_async(handle, interval, write_func):
    loop = get_event_loop()

    while not handle.closed:
        try:
            line = await loop.run_in_executor(None, readline, handle)
        except NoNewData:
            await async_sleep(interval)

        # try没有出现异常时，else会执行
        else:
            await write_func(line)


async def run_tasks(handles, interval, output_path):
    """
        将 tail_file改为 tail_async协程之后，就可以把原来为了执行该操作而使用的 get_event_loop 与 run_in_executor
        从顶层函数 run_tasks里往下推，把他们放到调用栈的下一层。这里，他们下沉到了 tail_async里面，
        于是我们可以把顶层函数中的那两条对应语句删掉。
    """
    with open(output_path, 'wb') as output:
        async def write_async(data):
            output.write(data)

        tasks = []
        for handle in handles:
            coro = tail_async(handle, interval, write_async)
            task = create_task(coro)
            tasks.append(task)

        await gather(*tasks)


# 运行, asyncio.run
# run(run_tasks_mixed(handles, 0.1, 'data/output.log'))
"""
    继续往下，tail_async中依赖的 readline函数也按照前面讲的三步转换为协程，
    但是，readline本身就是执行多项阻塞式的文件I/O操作，所以，好像没必要移植。
"""

# # ================================ [ 移植方式二：从下到上 ] ================================
"""
步骤：
    1）为要移植的每个末端函数（leaf function）都创建一个对应的异步协程版本
    2）修改现有的同步版本函数，把它原来执行的那些实际操作全部都拿掉，让它只通过事件循环去调用第1步里创建的相应的协程
    3）沿着调用栈向上移动一层，针对这一层里的相关函数制作对应的异步版本，并让这些异步版本函数去调用第一步里创建的相应协程
    4）把第2步中纯粹为了封装协程而设的同步版本删掉，因为上一层现在调用的是下一层里的异步版本函数，
       这些同步版本现在已经用不到了
"""

# 改进
# 决定让最底层的 readline依然执行阻塞式I/O，所以，现在的末端函数实际上是 tail_file，
# 按照步骤1，给它编写配套的协程版本, 也就是tail_async，已经有了

# 执行第2步，也就是把 tail_file的实际功能去掉，让它只负责将 tail_async封装起来，为了做到这一点，
# 我们必须想办法确保 tail_async协程能够运行完毕。
# 给每一个 tail_file工作线程都创建一套事件循环，并在事件循环上面调用 run_until_complete方法，
# 这个方法会阻塞当前线程，并推进事件循环去执行 coro所表示的 tail_async协程，直到该协程退出为止
